package com.atguigu.edu.realtime.util;

import com.atguigu.edu.realtime.bean.TransientSink;
import com.atguigu.edu.realtime.common.EduConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author Lec
 * @date 2022/8/24 16:18
 */

public class ClickHouseUtil {

    public static <T> SinkFunction<T> getClickHouseSinkFunc(String sql) {
        return JdbcSink.sink(sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        /*--------------------------泛型和反射的一些使用---------------------------*/
                        Class<?> tClass = t.getClass();

                        Field[] declaredFields = tClass.getDeclaredFields();

                        int jump = 0;

                        for (int i = 0; i < declaredFields.length; i++) {
                            declaredFields[i].setAccessible(true);
                            TransientSink annotation = declaredFields[i].getAnnotation(TransientSink.class);
                            if (annotation != null) {
                                jump++;
                            } else {
                                /*--------------------------get---------------------------*/
                                preparedStatement.setObject(i + 1 - jump, declaredFields[i].get(t));
                            }
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(10)
                        .withBatchIntervalMs(100L)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(EduConfig.CLICKHOUSE_URL)
                        .withDriverName(EduConfig.CLICKHOUSE_DRIVER)
                        /*clickhouse没有账号和密码*/
                        .build()
        );

    }

    /**
     *
     * @param table clickhouse中 表名
     * @param clazz 元素类型
     * @param <T> 泛型
     * @throws Exception 异常
     *
     *
     * keywordBeanDataStream.addSink(ClickHouseUtil.<KeywordBean>getSinkFunc("clickHouseTable", KeywordBean.class));
     */

    public static <T> SinkFunction<T> getSinkFunc(String table, Class<T> clazz) throws Exception {
        Field[] declaredFields = clazz.getDeclaredFields();
        int usefulFieldSize = 0;
        for (Field declaredField : declaredFields) {
            declaredField.setAccessible(true);
            TransientSink annotation = declaredField.getAnnotation(TransientSink.class);
            if (null == annotation) usefulFieldSize++;
        }

        if (usefulFieldSize == 0) {
            throw new Exception("不能往表中插入空白记录");
        }

        // insert into table values (?,?,?,?)
        StringBuilder sb = new StringBuilder("insert into " + table + " values (");
        for (int i = 0; i < usefulFieldSize; i++) {
            if (i != usefulFieldSize - 1) {
                sb.append("? , ");
            } else {
                sb.append("? )");
            }
        }

        return JdbcSink.<T>sink(sb.toString(),
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T keywordBean) throws SQLException {
                        Field[] fields = keywordBean.getClass().getDeclaredFields();
                        int skip = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            field.setAccessible(true);
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (null != annotation) {
                                skip++;
                                continue;
                            }
                            preparedStatement.setObject(i - skip + 1, field.get(keywordBean));
                        }
                    }
                },
                JdbcExecutionOptions
                        .builder()
                        .withBatchSize(1500)
                        .withMaxRetries(2)
                        .withBatchIntervalMs(5000)
                        .build(),
                //实例化内部类
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withUrl(EduConfig.CLICKHOUSE_URL)
                        .withDriverName(EduConfig.CLICKHOUSE_DRIVER)
                        .build());
    }
}
